package DianShang_2024.ds_01.tzgc


import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.functions.{asc, col, desc, sum, when}

import java.util.Properties

object tzgc01 {
  def main(args: Array[String]): Unit = {
    /*
          1、根据Hive的dwd库中相关表或MySQL中shtd_store中相关表（order_detail、sku_info），计算出与用户id为6708的用户所购买相同商品种类
          最多的前10位用户（只考虑他俩购买过多少种相同的商品，不考虑相同的商品买了多少次），将10位用户id进行输出，输出格式如下，将结果截图粘贴至客
          户端桌面【Release\任务C提交结果.docx】中对应的任务序号下；
    结果格式如下：
    -------------------相同种类前10的id结果展示为：--------------------
    1,2,901,4,5,21,32,91,14,52
     */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("第一套卷子特征工程01")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    //  导入隐式转换解决dataframe转化为dataset后的数据类型不匹配的问题(?:Encoder[Int])
    import spark.implicits._

    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")


    //  先拿到表格的数据作为数据源
    val order_detail=spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_detail",connect)
    val order_info=spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_info",connect)

    //  根据order_id连接起来,查出每个用户购买的商品去重
    val data=order_detail.join(order_info,order_detail("order_id")===order_info("id"))
      .select("user_id","sku_id")
      .distinct()

    //  6708用户所购买过的商品
    val user6708_sku_id:Array[Int]=data.filter(col("user_id")===6708).select("sku_id").map(
//      答案里面这一步是这样的:(_:Row) (0).toString.toInt
      (row:Row) => row(0).toString.toInt
    ).collect()

    //  :_* 语法是用来将数组展开为参数列表。
    //  isin()判断是否在这个区间里面
      val user_ids:String=data
        .withColumn(
          "cos",
          when(col("sku_id").isin(user6708_sku_id:_*),0.1).otherwise(0.0)
        )
      .groupBy("user_id")
      .agg(sum("cos").as("same"))
      .filter(col("user_id")!==6708)
      .orderBy(desc("same"),asc("user_id"))         //  这里的asc也是排序的一种规则，是从小到大的排序规则，也就是当same的值相同时，会按照user_id的从小到大排序
      .limit(10)
      .map(
        (row:Row) => row(0).toString
      )
      .collect()
      .mkString(",")

    println("---------------------------相同种类前10的id结果展示为:-------------------------------")
    print(user_ids)


    spark.close()
  }

}
